package com.bw.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


public class BaseLogApp {
    public static void main(String[] args) throws Exception {

        //todo 1.初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //todo 1.设置并行度
        env.setParallelism(1);

//      todo 3.从kafka读取主流数据
        String topic = "topic_log";

        String groupId = "base_log_consumer";

        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
//        source.print();

        //todo 4.数据清洗 转换结构
        //4.1定义错误侧输出流
//        OutputTag<String> dirtyStreamTag = new OutputTag<String>("dirtyStream") {
//        };

        SingleOutputStreamOperator<String> cleanedStream = source.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                try {
                    collector.collect(s);
                } catch (Exception e) {
//                    context.output(dirtyStreamTag,s);
                    System.out.println(s);
                }
            }
        });

//        cleanedStream.print();

        //4.2 将脏数据写出到kafka指定主题
//        DataStream<String> dirtyStream = cleanedStream.getSideOutput(dirtyStreamTag);

//        String dirtyTopic = "dirty_data";
//        dirtyStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(dirtyTopic));
//        dirtyStream.print("脏数据:");
        // 4.3 转换主流数据结构 jsonStr -> jsonObj

        SingleOutputStreamOperator<JSONObject> map = cleanedStream.map(JSON::parseObject);


        // TODO 5. 新老访客状态标记修复
        // 5.1 按照 mid 对数据进行分组
        KeyedStream<JSONObject, String> keyedStream = map.keyBy(o -> o.getJSONObject("common").getString("mid"));


        // 5.2 新老访客状态标记修复

        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

            //创建状态   给相同mid共享   容器里面只存日期   value  list   map
            ValueState<String> firstViewDtState;

            //open  ： 首先走open方法
            @Override
            public void open(Configuration param)throws Exception{
                super.open(param);

                firstViewDtState=getRuntimeContext().getState(new ValueStateDescriptor<String>("firstViewDtState",String.class));
            }



            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                String isNew = jsonObject.getJSONObject("common").getString("is_new");

                //状态时间
                String firstValue = firstViewDtState.value();


                //获取当前数据的时间
                Long ts = jsonObject.getLong("ts");

                //把时间转换成年月日
                String toDate = DateFormatUtil.toDate(ts);

                if ("1".equals(isNew)){
                    if (firstValue == null){
                        firstViewDtState.update(toDate);
                    }else {
                        if (!firstValue.equals(toDate)){
                            isNew = "0";
                            jsonObject.getJSONObject("common").put("is_new",isNew);
                        }
                    }
                }else {
                    if (firstValue == null){
                        String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);

                        firstViewDtState.update(yesterday);
                    }
                }
                collector.collect(jsonObject);

            }
        });

        // TODO 6. 分流 page  start     err
        // 6.1 定义启动、曝光、动作、错误侧输出流
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {};

        // 6.2 分流
        SingleOutputStreamOperator<String> separatedStream = fixedStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                // todo 6.2.1 收集错误数据

                JSONObject err = jsonObject.getJSONObject("err");

                if (err != null) {
                    context.output(errorTag, jsonObject.toJSONString());
                }

                // 剔除 "err" 字段
                jsonObject.remove("err");

                // todo 6.2.2 收集启动数据
                JSONObject start = jsonObject.getJSONObject("start");
                if (start != null){
                    context.output(startTag,jsonObject.toJSONString());
                }else {
                    //获取page字段
                    JSONObject page = jsonObject.getJSONObject("page");
                    //获取common字段
                    JSONObject common = jsonObject.getJSONObject("common");
                    //获取ts字段
                    Long ts = jsonObject.getLong("ts");



                    // todo 6.2.3 收集曝光数据[{},{},{}]
                    JSONArray displays = jsonObject.getJSONArray("displays");
                    if (displays != null){
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            //创建空的json对象
                            JSONObject displayJo = new JSONObject();

                            displayJo.put("display",display);
                            displayJo.put("common",common);
                            displayJo.put("page",page);
                            displayJo.put("ts",ts);

                            context.output(displayTag,displayJo.toJSONString());
                        }
                    }

                    // 6.2.4 收集动作数据
                    JSONArray actions = jsonObject.getJSONArray("actions");

                    if (actions != null){
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            JSONObject actionJo = new JSONObject();
                            actionJo.put("action",action);
                            actionJo.put("common",common);
                            actionJo.put("page",page);
                            actionJo.put("ts",ts);

                            context.output(actionTag,actionJo.toJSONString());
                        }
                    }

                    // 6.2.5 收集页面数据
                    jsonObject.remove("displays");
                    jsonObject.remove("actions");

                    collector.collect(jsonObject.toJSONString());

                }
            }
        });

//        separatedStream.print("page>>>");
//        separatedStream.getSideOutput(startTag).print("start>>>");
//        separatedStream.getSideOutput(displayTag).print("displayTag>>>");
//        separatedStream.getSideOutput(actionTag).print("actionTag>>>");
//        separatedStream.getSideOutput(errorTag).print("errorTag>>>");


        // TODO 7. 将数据输出到 Kafka 的不同主题
        // 7.1 提取各侧输出流
        DataStream<String> startDs = separatedStream.getSideOutput(startTag);
        DataStream<String> displayDs = separatedStream.getSideOutput(displayTag);
        DataStream<String> actionDs = separatedStream.getSideOutput(actionTag);
        DataStream<String> errorDs = separatedStream.getSideOutput(errorTag);

        // 7.2 定义不同日志输出到 Kafka 的主题名称
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";

        separatedStream.print("page>>>>>>>>>>>");
        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDs.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDs.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDs.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDs.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));


        env.execute();
    }
}
